package org.geektimes.reactive.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class DefaultSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;

    //当前请求数量
    private int count = 0;

    //最大请求数量
    private long maxRequest = 2;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
    }

    @Override
    public void onNext(Object o) {

        //1.先校验当前的一些状态是否合法
        if (count == maxRequest) { // 当到达数据阈值时，取消 Publisher 给当前 Subscriber 发送数据
            subscription.cancel();
            System.out.println("本次数据达到最大限制,提前结束：" + o);
            return;
        }
        try{
            //2.处理相关的真实逻辑
            System.out.println("收到数据：" + o);
            doExecute(o);
            //3.执行完成
            this.onComplete();
        }catch (Exception e){
            this.onError(e);
        }finally {
            //4.当执行成功时计数
            //不管当前本次是否成功或失败执行次数+1;
            count++;
        }
    }

    /**
     * 做真实的业务处理
     * @param o
     */
    private void doExecute(Object o) {
        //可以交给当前类和子类完成自己的业务逻辑
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("遇到异常：" + t);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
